package com.kafka

/**
  * Created by
  */
object StreamAppUserData {

  def main(args: Array[String]) {

/*
    val Array(file,brokers, group, topics) = args

    val topicSet = topics.split(",").toSet
    val kafkaParams = Map[String, String](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG  -> brokers,
      ConsumerConfig.GROUP_ID_CONFIG -> group
    )
    val km = new KafkaManager(kafkaParams)
    val properties: util.List[String] = JsonUtil.readProperties(file)
    val sparkConf = new SparkConf().set("spark.streaming.stopGracefullyOnShutdown","true") //.setAppName(file).setMaster("local[2]")

    val sc = new SparkContext(sparkConf)
    val ssc = new StreamingContext(sc, Seconds(60))
    val hiveContext = new HiveContext(sc)
    import hiveContext.implicits._
    hiveContext.setConf("hive.exec.dynamic.partition", "true")
    hiveContext.setConf("hive.exec.dynamic.partition.mode", "nonstrict")
    //    val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap
    //    val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2)
    val lines = km.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicSet)
    lines.foreachRDD(rdd => {
      if (!rdd.isEmpty()) {
        // 先处理消息
        val words = rdd.map(_._2).flatMap(JsonUtil.JsonToList(_,properties).toArray)
        //        words.collect().foreach(println)
        val time = new SimpleDateFormat("yyyy-MM-dd").format(new Date())
        import com.lava.kafka.Dome.Appuserdata
        import com.lava.kafka.Dome.ErrorData
        words.map(x=>x.toString.split("\001")).filter(x=>x.length==13).map(x=>Appuserdata(x(0),x(1),x(2),x(3),x(4),x(5),x(6),x(7),x(8),x(9),x(10),x(11),x(12),time)).
          toDF().write.partitionBy("date").mode(SaveMode.Append).saveAsTable("tb_lava_appuserdata")
        words.map(x=>x.toString.split("\001")).filter(x=>x.length!=13).map(x=>ErrorData("appuserdata",x.mkString("\001"),time))
          .toDF().write.partitionBy("date").mode(SaveMode.Append).saveAsTable("tb_lava_errordata")
        // 再更新offsets
        km.updateZKOffsets(rdd)
      }
    })

    ssc.start()
    ssc.awaitTermination()
    */
  }


}
